package concurency_io

import (
	"fmt"
	"io"
	"time"
)

type writerType = io.Writer
type readerType = io.Reader
type nameType = string

var (
	defaultErrTimeout = 3 // 秒
)

// 分发IO数据的实例
type TreeIOReader struct {
	ind       map[nameType]int
	dstWriter []*TreeIOWriter
	src       readerType
	mainWrite writerType
}

// 各个接收冗余数据的子实例
type TreeIOWriter struct {
	name       nameType
	w          writerType
	errChan    chan error
	errTimeOut int // 错误发送超时的时间，秒
	pipeWriter *io.PipeWriter
	pipeReader *io.PipeReader
}

func (w *TreeIOWriter) ErrChan() chan error {
	return w.errChan
}

func (w *TreeIOWriter) sendErrAsync(err error) {
	go func() {
		// 可能出现内存泄漏，使用超时控制
		timeout := time.NewTimer(time.Duration(w.errTimeOut) * time.Second)
		defer timeout.Stop()
		if w.errChan != nil {
			select {
			case w.errChan <- err:
			case <-timeout.C:
			}
		}
	}()
}

// 一次copy里会调用成百上千次Read+Write
//
//	func (w *TreeIOWriter) writeAsync(data []byte) {
//		go func() {
//			defer func() {
//				if rec := recover(); rec != nil {
//					fmt.Println("panic: ", rec)
//				}
//			}()
//			_, err := w.w.Write(data)
//			if nil != err {
//				w.sendErrAsync(err)
//				return
//			}
//		}()
//	}
func (w *TreeIOWriter) writeAsync() {
	go func() {
		defer func() {
			if rec := recover(); rec != nil {
				fmt.Println("panic: ", rec)
			}
		}()
		_, err := io.Copy(w.w, w.pipeReader) // 此处调用是从channel中接收数据,会阻塞等待,直到channel关闭
		if nil != err && err != io.EOF {
			w.sendErrAsync(err)
			return
		}
	}()
}

func (w *TreeIOWriter) GetWrite() writerType {
	return w.w
}

func NewTreeIOWriter(name nameType, w writerType, echan chan error) *TreeIOWriter {
	pr, pw := io.Pipe()
	return &TreeIOWriter{
		name:       name,
		w:          w,
		errChan:    echan,
		errTimeOut: defaultErrTimeout,
		pipeWriter: pw,
		pipeReader: pr,
	}
}

func NewTreeIOReader(r readerType, w writerType, wList ...*TreeIOWriter) *TreeIOReader {
	m := make(map[nameType]int)
	for i, w := range wList {
		m[w.name] = i
	}

	return &TreeIOReader{
		ind:       m,
		src:       r,
		dstWriter: wList,
		mainWrite: w,
	}
}

// 从源头中直接读取出来,然后将内容迭代发送给各个下游
func (t *TreeIOReader) Read(p []byte) (n int, err error) {
	n, err = t.src.Read(p)
	if n > 0 {
		data := p[:n]
		for _, w := range t.dstWriter {
			if n, err := w.pipeWriter.Write(data); err != nil {
				return n, err
			}
		}

	}
	return
}

// 如果要并发,还是应该利用协程,并且使用channel,会比较实时
func (w *TreeIOReader) sendToOtherAsync() error {
	for _, wr := range w.dstWriter {
		wr.writeAsync()
	}
	return nil
}

func (w *TreeIOReader) Start() error {
	w.sendToOtherAsync()
	io.Copy(w.mainWrite, w)
	return nil
}

func (w *TreeIOReader) Close() error {
	var err error
	for _, _w := range w.dstWriter {
		_w.pipeWriter.CloseWithError(err)
	}
	return nil
}
